﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Messaging;
using System.Runtime.InteropServices;
using System.Transactions;
using ShuttleBus.AdminMessages;
using ShuttleBus.Serialization;
using ShuttleBus.Subscriptions;

namespace ShuttleBus.Transports.Msmq
{
    internal class MsmqSubscriptionRepository : ISubscriptionRepository, IHandle<AddSubscriptionMessage>
    {
        private readonly MessageQueue _queue;
        private readonly ICreateMessages _messageFactory;
        private readonly ISerializer _serializer;
        private readonly ICreateStreams _factory;
        private readonly IReadStreams _reader;
        private readonly ConcurrentDictionary<string, Subscription> _subscriptions = new ConcurrentDictionary<string, Subscription>();

        public MsmqSubscriptionRepository(MessageQueue queue, ICreateMessages messageFactory, ISerializer serializer, ICreateStreams factory, IReadStreams reader)
        {
            _queue = queue;
            _messageFactory = messageFactory;
            _serializer = serializer;
            _factory = factory;
            _reader = reader;
        }

        internal void Initialize()
        {
            var enumerator = _queue.GetMessageEnumerator2();
            while (enumerator.MoveNext(TimeSpan.FromMilliseconds(0)))
            {
                var current = enumerator.Current;
                if (current == null)
                {
                    continue;
                }

                var msmqSubscription = _queue.Formatter.Read(current);

                /*if (msmqSubscription != null)
                {
                    var subscription = new Subscription(new Uri(msmqSubscription.Sender), Type.GetType(msmqSubscription.MessageType));
                    _subscriptions.AddOrUpdate(msmqSubscription.MessageType, key => subscription, (key, u) => subscription);
                }*/
            }
        }

        public void Add(AddSubscriptionMessage subscription)
        {
            var sub = subscription.ToSubscription();

            if (_subscriptions.TryAdd(sub.Location.ToString(), sub))
            {
                var message = subscription.OriginalMessage as System.Messaging.Message;
                IntPtr queueHandle = IntPtr.Zero;

                var subQueue = _queue.FormatName + ";subscriptions";

                int error = NativeMethods.MQOpenQueue(subQueue, NativeMethods.MQ_MOVE_ACCESS, NativeMethods.MQ_DENY_NONE, ref queueHandle);
                if (error == 0)
                {
                    Transaction current = Transaction.Current;
                    IDtcTransaction transaction = null;
                    if (current != null && _queue.Transactional)
                    {
                        transaction = TransactionInterop.GetDtcTransaction(current);
                    }
                    error = NativeMethods.MQMoveMessage(_queue.ReadHandle, queueHandle, message.LookupId, transaction);
                }
                error = NativeMethods.MQCloseQueue(queueHandle);
            }
        }

        public IEnumerable<Subscription> GetSubscribersOf(Message message)
        {
            var name = message.GetType().AssemblyQualifiedName;
            Subscription subscription;
            if (_subscriptions.TryGetValue(name, out subscription))
            {
                yield return subscription;
            }
        }

        public void Handle(AddSubscriptionMessage message)
        {
            var subscription = new Subscription(new Uri(message.Sender), Type.GetType(message.MessageType));
            _subscriptions.AddOrUpdate(message.MessageType, key => subscription, (key, u) => subscription);
        }
    }

    public static class NativeMethods
    {
        public const int MQ_MOVE_ACCESS = 4;
        public const int MQ_DENY_NONE = 0;
        public const int MQ_ADMIN_ACCESS = 0x0080;

        [DllImport("mqrt.dll", CharSet = CharSet.Unicode)]
        public static extern int MQOpenQueue(string formatName, int access, int shareMode, ref IntPtr hQueue);

        [DllImport("mqrt.dll")]
        public static extern int MQCloseQueue(IntPtr queue);

        [DllImport("mqrt.dll")]
        public static extern int MQMoveMessage(IntPtr sourceQueue, IntPtr targetQueue, long lookupID, IDtcTransaction transaction);

        [DllImport("mqrt.dll")]
        internal static extern int MQMgmtGetInfo([MarshalAs(UnmanagedType.BStr)]string computerName, [MarshalAs(UnmanagedType.BStr)]string objectName, ref MQMGMTPROPS mgmtProps);

        public const byte VT_NULL = 1;
        public const byte VT_UI4 = 19;
        public const int PROPID_MGMT_QUEUE_MESSAGE_COUNT = 7;

        //size must be 16
        [StructLayout(LayoutKind.Sequential)]
        internal struct MQPROPVariant
        {
            public byte vt;       //0
            public byte spacer;   //1
            public short spacer2; //2
            public int spacer3;   //4
            public uint ulVal;    //8
            public int spacer4;   //12
        }

        //size must be 16 in x86 and 28 in x64
        [StructLayout(LayoutKind.Sequential)]
        internal struct MQMGMTPROPS
        {
            public uint cProp;
            public IntPtr aPropID;
            public IntPtr aPropVar;
            public IntPtr status;

        }
    }
}